package com.supreme.commonrocketmq.consumer;

import com.alibaba.fastjson.JSONObject;
import com.supreme.commonrocketmq.config.props.RocketmqProps;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

@Slf4j
//@Component
public class MysqlBinlogConsumer {

    @Resource
    private RocketmqProps rocketmqProps;

    @PostConstruct
    private void initConsumer() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketmqProps.getProducerNameSrv());
        defaultMQPushConsumer.setConsumerGroup(rocketmqProps.getMysqlBinlog().getGroup());
        try {
            defaultMQPushConsumer.subscribe(rocketmqProps.getMysqlBinlog().getName(), "*");
            defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExtList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    messageExtList.forEach(messageExt -> {
                        byte[] bodyBytes = messageExt.getBody();
                        JSONObject bodyJson = JSONObject.parseObject(bodyBytes, JSONObject.class);
                        System.out.println("bodyJson = " + bodyJson);

                        /*String database = bodyJson.getString("database");
                        String table = bodyJson.getString("table");
                        String type = bodyJson.getString("type");
                        if (StringUtils.equals(database, "supreme_dev") && StringUtils.equals(table, "test")) {
                            ESTest[] dataList = bodyJson.getObject("data", ESTest[].class);
                            if (dataList.length > 0) {
                                ESTest esTest = dataList[0];
                                try {
                                    String documentId = String.valueOf(esTest.getId());
                                    if (StringUtils.equals(type, "INSERT")) {
                                        elasticsearchService.createDocument("mysql_test_index", esTest, documentId);
                                    } else if (StringUtils.equals(type, "UPDATE")) {
                                        elasticsearchService.updateDocument("mysql_test_index", documentId, esTest);
                                    }
                                } catch (IOException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        }*/
                    });

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            log.error(e.getMessage(), e);
        }
    }
}
